package cn._51doit.day07;

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.state.ListState;
import org.apache.flink.api.common.state.ListStateDescriptor;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;

/**
 * @create: 2021-10-23 21:23
 * @author: 今晚打脑斧先森
 * @program: ListStateDemo
 * @Description:
 *  将同一个用户的 行为 保存起来
 **/
public class ListStateDemo {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.enableCheckpointing(10000);
        DataStreamSource<String> lines = env.socketTextStream("doit01", 8888);

        SingleOutputStreamOperator<Tuple2<String, String>> map1 = lines.map(new MapFunction<String, Tuple2<String, String>>() {
            @Override
            public Tuple2<String, String> map(String value) throws Exception {
                String[] split = value.split(",");
                return Tuple2.of(split[0], split[1]);
            }
        });
        KeyedStream<Tuple2<String, String>, String> keyedStream = map1.keyBy(t -> t.f0);
        SingleOutputStreamOperator<Tuple2<String, List<String>>> res = keyedStream.process(new ProcessFunction<Tuple2<String, String>, Tuple2<String, List<String>>>() { //返回的是行为集合,所有定义一个集合

            //使用transient修饰,这样客户端就没法更改它了,因为它就不参与反序列化和序列化了
            //因为即使客户端更改了,数据也不会随着字节码文件流到堆内存中
            private transient ListState<String> listState;

            @Override
            public void open(Configuration parameters) throws Exception {
                //定义描述器,描述器里面存储的是string类型数据,要全部存储起来,所以要用list集合
                ListStateDescriptor<String> listStateDescriptor = new ListStateDescriptor<>("行为", String.class);
                listState = getRuntimeContext().getListState(listStateDescriptor);
            }

            @Override
            public void processElement(Tuple2<String, String> value, Context ctx, Collector<Tuple2<String, List<String>>> out) throws Exception {
                String userName = value.f0; //用户姓名
                String userBehavier = value.f1;//用户的行为

                //直接将数据add到行为状态里面
                listState.add(userBehavier);
                //因为listState底层就是用的集合,所以可以直接强转
                ArrayList<String> events = (ArrayList<String>) listState.get();
                //输出数据
                out.collect(Tuple2.of(userName, events));
            }
        });
        res.print();
        env.execute();
    }
}
